package org.zjt.rxjava.lession.first;

import io.reactivex.*;
import io.reactivex.disposables.Disposable;
import lombok.SneakyThrows;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.PrintStream;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

/**
 * RXjava:
 * <p>
 * 优点：
 * 1、当订阅者发生异常不能执行时候，上游就不会执行。
 * 2、面向的是过程，当业务需求发送变化时候，可以对过程实现操作，实现解耦。
 */
public class ObservLearning {

    private static final Logger LOG = LoggerFactory.getLogger(ObservLearning.class);

    private static final PrintStream origin = System.out;

    static {
        System.setOut(System.err);
    }

    private static final PrintStream out = System.out;

    static {
        System.setOut(origin);
    }


    @SneakyThrows
    public static void main(String[] args) throws InterruptedException {

        Flowable.just("Hello world").subscribe(LOG::info);

        /**
         * 当订阅者出错不能订阅时候，当前观察者不会执行。
         */
        Observable<String> observable = Observable.create(a -> {
            a.onNext("Hell,o,wor,ld");
            a.onNext("H2ell,o,wor,ld");
            a.onComplete();
            //int i =1/0;
            // TODO: 2017/5/18  onError()函数向订阅者发送一个异常。
            //a.onError(new Exception("zhangsan"));
        });


        // TODO: 2017/5/18 函数式编程： 调用subscribe(Consumer<? super T> onNext)
        //                             Consumer<? super T>则是一个唯一函数的接口，用来传输函数的引用。
        observable
                .flatMap(a -> Observable.fromArray(a.split(",")))
                .map(a -> a.contains("o") ? a.toUpperCase() : a)
                .subscribe(LOG::error);

        observable.subscribe(LOG::info);

        observable.subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable disposable) {
                LOG.info("onSubscribe():{}", disposable.toString());
            }

            @Override
            public void onNext(String s) {
                LOG.info("onNext():{}", s);
            }

            @Override
            public void onError(Throwable throwable) {
                LOG.error("onSubscribe():{}", throwable.toString());
            }

            @Override
            public void onComplete() {
                LOG.info("onComplete():{}", "完成");
            }
        });


        ArrayList<String> objects = new ArrayList<String>();
        objects.add("aaaa");
        objects.add("bbssb");
        objects.add("dddd");
        Observable<String> array = Observable.fromArray(objects).flatMap(a -> Observable.fromArray(a.toArray())).map(a -> a.toString());
        array.subscribe(LOG::info);

        out.println("打印just：");
        Observable.just("hello just!", "hello just2!").map(String::toUpperCase).subscribe(LOG::info);


        out.println("打印zip：");
        Observable.zip(array, observable, (a, b) -> a.toString() + b.toString()).subscribe(LOG::info);


        //  使用defer，有观察者订阅时才创建Observable，并且为每个观察者创建一个新的
        Observable.defer(() -> Observable.just("zhang san")).subscribe(LOG::info);


        //interval 循环定时发送
        Observable.interval(1, TimeUnit.SECONDS).subscribe(out::println);//每隔一秒发送一次 发送一个执行次数
        Thread.sleep(5000);


        //range 数组发送
        Observable.range(10, 5).timeInterval(TimeUnit.SECONDS).subscribe(out::println);


        //timer 3秒后发射一个值  0
        Observable.timer(2, TimeUnit.SECONDS).subscribe(out::println);
        Thread.sleep(10000);


        //repeat 重复发射3次
        Observable.just("repeatObservable").repeat(3).subscribe(out::println);





    }
}



